package com.biz.aoao.mysql.mysqlbinlogsync.core;

import com.biz.aoao.mysql.mysqlbinlogsync.MySqlBinLogSyncConstants;
import com.biz.aoao.mysql.mysqlbinlogsync.config.MySqlHost;
import com.biz.aoao.mysql.mysqlbinlogsync.config.MySqlHostLists;
import com.biz.aoao.mysql.mysqlbinlogsync.config.TalbeConfigInfo;
import com.biz.aoao.mysql.mysqlbinlogsync.extra.DbUtils;
import com.biz.aoao.mysql.mysqlbinlogsync.listener.CustomerEventListener;
import com.biz.aoao.mysql.mysqlbinlogsync.listener.IMysqlDataListener;
import com.github.shyiko.mysql.binlog.BinaryLogClient;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.SmartInitializingSingleton;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

import java.io.IOException;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
import java.util.stream.Stream;

@Component
@Slf4j
public class BinLogProcessor implements SmartInitializingSingleton, ApplicationContextAware {

    private ApplicationContext context;
    @Override
    public void setApplicationContext(ApplicationContext context) throws BeansException {
        this.context=context;
    }

    @Autowired
    private MySqlHostLists mySqlHostLists;

    @Autowired
    private CustomerEventListener customerEventListener;
    /**
     * 当所有单例 bean 都初始化完成以后， 容器会回调该接口的方
     */
    @Override
    public void afterSingletonsInstantiated() {

        /*获取Spring容器中所有实现了IMysqlDataListener的对象*/
        Map<String, IMysqlDataListener> mysqlDateListeners = context.getBeansOfType(IMysqlDataListener.class);

        Map<Long, List<MysqlDataListenerData>> listenerMaps =
                mysqlDateListeners.values().stream().map(MysqlDataListenerData::new).collect(Collectors.groupingBy(MysqlDataListenerData::getServerid));
        //根据name找到对应的mysql配置信息
        if(listenerMaps!=null&&listenerMaps.size()>0){
            listenerMaps.forEach((k,v)->{
                //启动线程来监听binlog日志
                MySqlHost mysqlHostByName = mySqlHostLists.getMysqlHostByServerId(k);
                if(mysqlHostByName!=null){
                    initTableInfo(v,mysqlHostByName);
                    new Thread(){
                        @Override
                        public void run() {

                            BinaryLogClient client = new BinaryLogClient(mysqlHostByName.getHost(), mysqlHostByName.getPort(), mysqlHostByName.getUsername(), mysqlHostByName.getPassword());

                            if(mysqlHostByName.getFilename()!=null&&!"".equals(mysqlHostByName.getFilename())){
                                client.setBinlogFilename(mysqlHostByName.getFilename());
                            }
                            if(mysqlHostByName.getPosition()!=null){
                               client.setBinlogPosition(mysqlHostByName.getPosition());
                            }

                            client.registerEventListener(customerEventListener);

                            try {
                                client.connect();
                            } catch (IOException e) {
                                log.error("启动MysqlBinLog监听出错",e);
                            }
                        }
                    }.start();
                }

            });
        }



    }


    /**
     * 初始化表信息到内存中
     */
    private void initTableInfo(List<MysqlDataListenerData> datas,MySqlHost mysqlHostByName){

        if(CollectionUtils.isEmpty(datas)){
            return;
        }
        datas.forEach(data->{
            String[] columns = DbUtils.getColumns(mysqlHostByName, data.getDatabase(), data.getTable());
            String key=data.getDatabase()+MySqlBinLogSyncConstants.DATEBASE_TABLE_FLEX+data.getTable();
            TalbeConfigInfo talbeConfigInfo=new TalbeConfigInfo();
            talbeConfigInfo.setColumns(columns);
            talbeConfigInfo.setDatabase(data.getDatabase());
            talbeConfigInfo.setTableName(data.getTable());
            talbeConfigInfo.setEntityClass(data.getEntityClass());
            talbeConfigInfo.setListener(data.getListener());
            talbeConfigInfo.setServerid(data.getServerid());
            MySqlBinLogSyncConstants.TABLE_INFO_MAPPING.put(key,talbeConfigInfo);
        });


    }
}
